package com.hzgj.bcl.soa.interceptor;

import com.google.common.collect.Maps;

import com.hzgj.bcl.soa.Event;
import com.hzgj.bcl.soa.rpc.message.Message;
import com.hzgj.bcl.soa.spi.MessageHandleFactory;
import com.hzgj.bcl.soa.util.Constants;
import com.hzgj.bcl.soa.util.ServiceLoaderUtil;
import com.hzgj.bcl.util.lang.StrKit;

import java.util.Iterator;
import java.util.Map;
import java.util.ServiceLoader;

import lombok.Setter;

import static com.hzgj.bcl.soa.util.Constants.MESSAGE_HEADER_PROTOCOL;

/**
 * Created by dell on 2015/3/31.
 */
public class SerializationInterceptor extends EventFliterInterceptor {

    /**
     * key:id
     */
    private static Map<Byte, Serialization> cacheById = Maps.newConcurrentMap();
    private static Map<String, Serialization> cacheByProtocol = Maps.newConcurrentMap();

    static {
        ServiceLoader<Serialization> serviceLoader = ServiceLoader.load(Serialization.class);
        if (serviceLoader != null) {
            Iterator<Serialization> it = serviceLoader.iterator();
            while (it.hasNext()) {
                Serialization serialization = it.next();
                cacheById.put(serialization.getId(), serialization);
                cacheByProtocol.putIfAbsent(serialization.getName(), serialization);
            }
        }
    }

    @Setter
    Serialization serialization = new FastSerialization();

    @Override
    public Object down0(Event evt) {
        Message msg = (Message) evt.getArg();
        if (needSerialization(msg)) {
            Serialization serialization = getSerialization(msg);
            try {
                msg.setPayload(serialization.write(msg));
                msg.putHeader(Constants.MESSAGE_HEADER_SERIALIZATION_ID, serialization.getId());
            } catch (Exception e) {
                Event event = getSerializationErrorInfo(e, msg);
                if (event != null) {
                    evt = event;
                }
            }
        }
        return down_int.down(evt);
    }

    @Override
    public Object up0(Event evt) {
        Message msg = (Message) evt.getArg();
        if (needSerialization(msg)) {
            Serialization serialization = getSerialization(msg);
            try {
                msg.setPayload(serialization.read(msg));
            } catch (Exception e) {
                Event event = getSerializationErrorInfo(e, msg);
                if (event != null) {
                    evt = event;
                }
            }
        }
        return up_int.up(evt);
    }

    private Event getSerializationErrorInfo(Exception e, Message msg) {
        try {
            MessageHandleFactory factory = ServiceLoaderUtil.getMessageBuildFactory((String) msg.getHeader(MESSAGE_HEADER_PROTOCOL));
            Message message = factory.buildErrorResponse(e, msg.getHeaders());
            message.setPayload(serialization.write(message));
            message.putHeader(Constants.MESSAGE_TYPE, Message.EXC_RSP)
                    .putHeader(Constants.MESSAGE_HEADER_SERIALIZATION_ID, serialization.getId());
            return new Event(Event.MSG, message);
        } catch (Exception e1) {
        }
        return null;
    }

    public Serialization getSerialization(Message msg) {
        Byte id = (Byte) msg.getHeader(Constants.MESSAGE_HEADER_SERIALIZATION_ID);
        if (id != null) {
            serialization = findSerialization(id);
        } else {
            serialization = cacheByProtocol.get(msg.getHeader(MESSAGE_HEADER_PROTOCOL));
        }
        if (serialization == null) {
            throw new RuntimeException("can't init getSerialization by message " + msg);
        }
        return serialization;
    }

    private Serialization findSerialization(Byte id) {
        Serialization serialization = cacheById.get(id);
        if (serialization != null) {
            return serialization;
        }
        return new FastSerialization();
    }


    private boolean needSerialization(Message message) {
        Object header = message.getHeader(Constants.MESSAGE_HEADER_SERIALIZATION_BYPASS);
        boolean pass = header != null && (Boolean) header;
        return pass ? false : true;
    }
}
